package com.itc.bi.kafaka;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.itc.bi.constant.KafkaConstant;
import com.itc.bi.entity.*;
import com.itc.bi.handler.dataSet.PythonCode;
import com.itc.bi.kafaka.DTO.DataSetKafkaDTO;
import com.itc.bi.service.*;
import com.itc.bi.service.impl.ClickHouseJDBCService;
import com.itc.common.core.exception.ServiceException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;

/**
 * @BelongsProject: itc-airport-cloud
 * @BelongsPackage: com.itc.bi.kafaka
 * @Author: hecaiy
 * @CreateTime: 2024-07-04  10:43
 * @Description: TODO python数据集创建操作
 * @Version: 1.0
 */
@Component
@Slf4j
@Service
public class DataSetPythonKafkaListener {
    @Autowired
    private DirectoryService directoryService;
    @Autowired
    private ClickHouseJDBCService clickHouseJDBCService;
    @Autowired
    private DataSetColumnService dataSetColumnService;
    @Autowired
    private DirectDataSetService directDataSetService;
    @Autowired
    private PythonCode pythonCode;
    @Autowired
    private XxlJobInfoService xxlJobInfoService;
    @Autowired
    private DataSynrecordService dataSynrecordService;
    @Autowired
    XxlJobGroupService groupService;
    @Value("${xxl.job.executor.appname}")
    private String jobName;

    /**
     * 1.python构建数据集
     * 2.创建主附表
     * 3.数据同步
     * 4.xxl-job任务调度
     * @param record
     */
    @KafkaListener(groupId = "bi_data_set_python",topics = KafkaConstant.KAFKA_TOPIC_DATA_SET_PYTHON)
    public void dataSetListener(ConsumerRecord record){
        String message = (String) record.value();
        DataSetKafkaDTO directoryParam = JSON.parseObject(message, DataSetKafkaDTO.class);
        if(ObjectUtil.isNotEmpty(directoryParam)){
            Long dtId = directoryParam.getDtId();
            DirectoryEntity directoryEntity = directoryService.getById(dtId);
            if(ObjectUtil.isEmpty(directoryEntity)){
                log.error("数据集信息为空");
                throw new ServiceException("数据集信息为空");
            }
            if(ObjectUtil.isNotEmpty(directoryEntity.getVersion()) && ObjectUtil.isNotEmpty(directoryEntity.getTableName())){
                log.error("数据集已存在");
                return ;
            }
            directoryEntity.setTableName(KafkaConstant.DATA_SET_TYPE_PYTHON_MASTER+dtId);
            directoryEntity.setVersion(KafkaConstant.DATA_SET_TYPE_PYTHON_SLAVE+dtId);
            //数仓数据表创建，数据同步
            List<DataSetColumnEntity> list = dataSetColumnService.list(Wrappers.<DataSetColumnEntity>lambdaQuery().eq(DataSetColumnEntity::getDataSetId, dtId));
            if(ObjectUtil.isEmpty(list)){
                log.error("数据集字段为空");
                throw new ServiceException("数据集字段为空");
            }
            clickHouseJDBCService.createTableByColumnList(directoryEntity.getTableName(), list,"");
            clickHouseJDBCService.createTableByColumnList(directoryEntity.getVersion(),list,"");
            DirectDataSetEntity dataSetEntity = directDataSetService.getById(dtId);
            List<Object> objects = pythonCode.previewData(dataSetEntity,null,null);
            if(ObjectUtil.isNotEmpty(objects)){
                clickHouseJDBCService.insertDataByTableNameObject(directoryEntity.getTableName(),objects);
            }
            this.directoryService.updateById(directoryEntity);
            //TODO  此处差创建xxl调度任务
            if(StringUtils.isEmpty(directoryParam.getCron())){
                return;
            }
            XxlJobGroupEntity groupEntity = groupService.getOne(Wrappers.<XxlJobGroupEntity>lambdaQuery().eq(XxlJobGroupEntity::getAppName, jobName).last("limit 1"));
            XxlJobInfoEntity xxlJobInfoEntity = new XxlJobInfoEntity();
            xxlJobInfoEntity.setJobGroup(groupEntity.getId());
            xxlJobInfoEntity.setJobDesc("数据集PYTHON调度任务"+directoryEntity.getName());
            xxlJobInfoEntity.setAddTime(new Date());
            xxlJobInfoEntity.setUpdateTime(new Date());
            xxlJobInfoEntity.setAuthor("lowcode_bi");
            xxlJobInfoEntity.setAlarmEmail("waytale@waytale.cn");
            xxlJobInfoEntity.setScheduleType("CRON");
            xxlJobInfoEntity.setScheduleConf(directoryParam.getCron());
            xxlJobInfoEntity.setMisfireStrategy("DO_NOTHING");
            xxlJobInfoEntity.setExecutorRouteStrategy("FIRST");
            xxlJobInfoEntity.setExecutorHandler("PythonHandler");
            JSONObject executorParam = new JSONObject();
            executorParam.put("dtId",dtId);
            xxlJobInfoEntity.setExecutorParam(JSON.toJSONString(executorParam));
            xxlJobInfoEntity.setExecutorFailRetryCount(0);
            xxlJobInfoEntity.setGlueType("BEAN");
            xxlJobInfoEntity.setExecutorBlockStrategy("SERIAL_EXECUTION");
            xxlJobInfoEntity.setExecutorTimeout(0);
            xxlJobInfoService.saveOrUpdateXxlJob(xxlJobInfoEntity);

            //记录创建的调度任务
            DataSynrecordEntity entity = dataSynrecordService.getOne(Wrappers.<DataSynrecordEntity>lambdaQuery()
                    .eq(DataSynrecordEntity::getDtId, dtId));
            DataSynrecordEntity dataSynrecordEntity = new DataSynrecordEntity();
            //若存在调度任务 停止原来的调度任务、删除原来的调度任务
            if(ObjectUtil.isNotEmpty(entity)){
                BeanUtil.copyProperties(entity,dataSynrecordEntity);
                xxlJobInfoService.stop(entity.getTaskId());
                xxlJobInfoService.removeJob(entity.getTaskId());
            }
            dataSynrecordEntity.setDtId(dtId);
            dataSynrecordEntity.setTableId(dtId);
            dataSynrecordEntity.setTaskId(xxlJobInfoEntity.getId());
            dataSynrecordEntity.setId(dataSynrecordEntity.getId());
            dataSynrecordEntity.setDirId(directoryEntity.getPid());
            dataSynrecordEntity.setTargetTableName(directoryEntity.getTableName());
            dataSynrecordEntity.setCron(directoryParam.getCron());
            dataSynrecordEntity.setIsExec(1);
            dataSynrecordEntity.setSynType(1);
            dataSynrecordService.saveOrUpdate(dataSynrecordEntity);

            //启动执行
            xxlJobInfoService.start(xxlJobInfoEntity.getId());
        }
    }
}
